package com.atguigu.flink.chapter11.window;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lzc
 * @Date 2022/11/1 09:34
 */
public class Flink01_Window_Group_1 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<WaterSensor> stream = env.fromElements(
            new WaterSensor("s1", 1000L, 10),
            new WaterSensor("s2", 1000L, 10),
            new WaterSensor("s1", 2000L, 20),
            new WaterSensor("s1", 3000L, 30),
            new WaterSensor("s1", 4000L, 40),
            new WaterSensor("s1", 5000L, 50),
            new WaterSensor("s1", 8000L, 50)
        ).assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((ws, ts) -> ws.getTs())
        );
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream, $("id"), $("ts"), $("vc"), $("et").rowtime());
        tEnv.createTemporaryView("sensor", table);
        
        /*tEnv.sqlQuery("select " +
                          " id, " +
                          " tumble_start(et, interval '5' second) stt, " +
                          " tumble_end(et, interval '5' second) edt, " +
                          " sum(vc) vc_sum " +
                          "from sensor " +
                          "group by id, tumble(et, interval '5' second)")
            .execute()
            .print();*/
    
        /*tEnv.sqlQuery("select " +
                          " id, " +
                          " hop_start(et, interval '2' second,interval '5' second ) stt, " +
                          " hop_end(et, interval '2' second,interval '5' second ) edt, " +
                          " sum(vc) vc_sum " +
                          "from sensor " +
                          "group by id, hop(et, interval '2' second,interval '5' second )")
            .execute()
            .print();*/
    
        tEnv.sqlQuery("select " +
                          " id, " +
                          " session_start(et, interval '2' second) stt, " +
                          " session_end(et, interval '2' second) edt, " +
                          " sum(vc) vc_sum " +
                          "from sensor " +
                          "group by id, session(et, interval '2' second)")
            .execute()
            .print();
        
 
        
    }
}
/*
分组窗口(group window)
    keyBy-> window-> agg
    
    0-5  s1  10
    0-5  s2  100
    5-10 s1  20
    ...
    select w.start, w.end id, sum(vc) from t group by id, w
    


over 窗口


 */